package cn.springboot.hadoop;

import java.io.IOException;
import java.io.StringReader;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.map.InverseMapper;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
import org.wltea.analyzer.core.IKSegmenter;
import org.wltea.analyzer.core.Lexeme;


/**
 * @author liufl E-mail:Huberier@allpyra.com
 * @version 创建时间：2017年9月25日 下午6:37:02
 * 拆词统计-出现次数
 */
public class HDFSWordSplit {
	/**
     * map实现分词
     * @author Administrator
     *
     */
    public static class TokenizerMapper extends Mapper<Object, Text, Text, IntWritable> {
        private static final IntWritable one = new IntWritable(1);
        private Text word = new Text();

        public void map(Object key, Text value, Mapper<Object, Text, Text, IntWritable>.Context context)
                throws IOException, InterruptedException {
            StringReader input = new StringReader(value.toString());
            IKSegmenter ikSeg = new IKSegmenter(input, true); // 智能分词
            for (Lexeme lexeme = ikSeg.next(); lexeme != null; lexeme = ikSeg.next()) {
                this.word.set(lexeme.getLexemeText());
                context.write(this.word, one);
            }
        }
    }

    /**
     * reduce实现分词累计
     * @author Administrator
     *
     */
    public static class IntSumReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
        private IntWritable result = new IntWritable();

        public void reduce(Text key, Iterable<IntWritable> values,
                Reducer<Text, IntWritable, Text, IntWritable>.Context context)
                throws IOException, InterruptedException {
            int sum = 0;
            for (IntWritable val : values) {
                sum += val.get();
            }
            this.result.set(sum);
            context.write(key, this.result);
        }
    }

    /**
     * 实现降序
     * 
     * @author Administrator
     *
     */
    @SuppressWarnings("rawtypes")
    private static class IntWritableDecreasingComparator extends IntWritable.Comparator {
		public int compare(WritableComparable a, WritableComparable b) {
            return -super.compare(a, b);
        }

        public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
            return -super.compare(b1, s1, l1, b2, s2, l2);
        }
    }

    public static void main(String[] args) throws Exception {
        Configuration conf = HDFSBase.config();
//        HDFSBase hdfs = new HDFSBase(conf); 
//        hdfs.mkdirs("/tmp/input");
//        hdfs.mkdirs("/tmp/output");
        //使用本地硬盘文件
        String inputFile = "/tmp/input/myext.dic"; // 输入文件-必须存在
        Path outDir = new Path("/tmp/output"); // 输出目录-可以不存在
        Path tempDir = new Path("/tmp" + System.currentTimeMillis()); // 临时目录
        //使用HDFS文件
//        String inputFile = HDFSBase.HDFSURL+"/tmp/input/person.txt"; // 输入文件-必须存在
//        Path outDir = new Path(HDFSBase.HDFSURL+"/tmp/output"); // 输出目录-可以不存在
//        Path tempDir = new Path(HDFSBase.HDFSURL+"/tmp" + System.currentTimeMillis()); // 临时目录

        HDFSWordSplit wordSplit = new HDFSWordSplit();
        // 第一个任务：分词
        System.out.println("start task...");
        Job job = Job.getInstance(conf, "word split");
        job.setJarByClass(wordSplit.getClass());
        job.setMapperClass(TokenizerMapper.class);
        job.setCombinerClass(IntSumReducer.class);
        job.setReducerClass(IntSumReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        FileInputFormat.addInputPath(job, new Path(inputFile));
        FileOutputFormat.setOutputPath(job, tempDir);
        System.out.println(">>>>>>>>>>>>"+FileInputFormat.getInputPaths(job)[0].toUri());

        // 第一个任务结束，输出作为第二个任务的输入，开始排序任务
        job.setOutputFormatClass(SequenceFileOutputFormat.class);
        if (job.waitForCompletion(true)) {
            System.out.println("start sort...");
            Job sortJob = Job.getInstance(conf, "word sort");
            sortJob.setJarByClass(wordSplit.getClass());
            sortJob.setMapperClass(InverseMapper.class);
            sortJob.setInputFormatClass(SequenceFileInputFormat.class);

            // 反转map键值，计算词频并降序
            sortJob.setMapOutputKeyClass(IntWritable.class);
            sortJob.setMapOutputValueClass(Text.class);
            sortJob.setSortComparatorClass(IntWritableDecreasingComparator.class);
            sortJob.setNumReduceTasks(1);

            // 输出到out目录文件
            sortJob.setOutputKeyClass(IntWritable.class);
            sortJob.setOutputValueClass(Text.class);
            FileInputFormat.addInputPath(sortJob, tempDir);

            // 如果已经有out目录，先删再创建
            FileSystem fileSystem = outDir.getFileSystem(conf);
            if (fileSystem.exists(outDir)) {
                fileSystem.delete(outDir, true);
            }
            FileOutputFormat.setOutputPath(sortJob, outDir);

            if (sortJob.waitForCompletion(true)) {
                System.out.println("finish and quit....");
                // 删掉临时目录
                fileSystem = tempDir.getFileSystem(conf);
                if (fileSystem.exists(tempDir)) {
                    fileSystem.delete(tempDir, true);
                }
                System.exit(0);
            }
        }
    }
}

